package job.antispam.graph_building.domain;

import java.io.IOException;

import mapred.base.TextMapper;

import org.apache.hadoop.io.Text;

import util.FeatureMap;
import util.UrlUtil;

public class DomainGraphGenerator_Mapper extends TextMapper {

    private static String delim;

    @Override
    public void map(Text key, Text value) throws IOException {
        String k = key.toString();
        if (k.endsWith(".")) {
            k = k.substring(0, k.length() - 1);
        }
        String domain = UrlUtil.getDomain(k);
        if (domain.isEmpty()) {
            this.incrCounter("Domain", "Error", 1L);
        }
        FeatureMap map = FeatureMap.createNewFeatureMap(delim);
        FeatureMap targets = FeatureMap.parse(value.toString(), delim);
        for (String target : targets.featureSet()) {
            String t = UrlUtil.getDomain(target);
            if (!domain.equals(t)) {
                map.addFeature(t, targets.get(target));
            }
        }
        this.output(new Text(domain), new Text(map.toString()));
    }

    @Override
    public void setUp() {
        delim = this.getConfig("delim");
    }

    @Override
    public void tearDown() throws IOException {
    }

}
